package com.wudl.core;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @version v1.0
 * @ProjectName Flinklearning
 * @ClassName TransformFilter
 * @Description TODO 流的过滤
 * @Author wudl
 * @Date 2020/11/5 10:26
 */

public class TransformFilter {


    /**
     * 函数中Filter 中过滤
     * 过滤：根据指定的规则将满足条件的(true) 的数据保留, 不瞒住条件的(false)  将丢弃
     * 返回值：DataStream
     */
    public static void main(String[] args) throws Exception {

        //1.获取上下文的环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.设置并行度
        env.setParallelism(1);
        //3.获取数据流
        DataStreamSource<String> SourceDs = env.socketTextStream("10.204.125.140", 8899);
        //4. 过滤数据流
        DataStream<String> filter = SourceDs.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                String[] split = value.split(",");
                return split[1].length() > 3;
            }
        });
        filter.print();
        env.execute();

    }


}
